package com.huatai.datacenter.service;

import com.alibaba.fastjson.JSONArray;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.huatai.datacenter.config.TopicConfig;
import com.huatai.datacenter.constant.Constants;
import com.huatai.datacenter.entity.ClusterEntity;
import com.huatai.datacenter.entity.KafkaRecord;
import com.huatai.datacenter.entity.TopicInfoEntity;
import com.huatai.datacenter.mapper.ClusterMapper;
import com.huatai.datacenter.mapper.TopicInfoMapper;
import com.huatai.datacenter.utils.KafkaProducers;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.time.Duration;
import java.time.LocalDateTime;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ExecutionException;

/**
 * @author Lion
 * @date 2023/3/24  9:27
 */
@Service
@Slf4j
public class TopicInfoService extends ServiceImpl<TopicInfoMapper, TopicInfoEntity> {

	@Autowired
	private TopicInfoMapper topicInfoMapper;

	@Autowired
	private KafkaAdminService kafkaAdminService;

	@Autowired
	private KafkaConsumerService kafkaConsumerService;

	@Autowired
	private ClusterMapper clusterMapper;

	/**
	 * @author Lion
	 * @description 根据集群id获取主题信息
	 * @param  clusterId
	 * @return java.util.List<com.huatai.datacenter.entity.TopicInfoEntity>
	 */
	public List<TopicInfoEntity> selectAllByClusterId(String clusterId) {
		return topicInfoMapper.getTopicsByCluster(clusterId);
	}
	/**
	 * @author Lion
	 * @description 获取所有主题信息
	 * @param
	 * @return java.util.List<com.huatai.datacenter.entity.TopicInfoEntity>
	 */
	public List<TopicInfoEntity> getTotalData(){
		return topicInfoMapper.getTopics();
	}

	/**
	 * @author Lion
	 * @description 删除主题
	 * @param topicName 主题名称
	 * @param clusterId 服务端id
	 * @return boolean
	 */
	boolean isDelete(String topicName, String clusterId) {
		try {
			Map<String,Object> map = new HashMap<>();
			map.put(Constants.KeyStr.TOPIC_NAME, topicName);
			map.put(Constants.KeyStr.CLUSTER_ID, clusterId);
			boolean topic = checkResult(topicInfoMapper.deleteByMap(map));
			map.remove(Constants.KeyStr.TOPIC_NAME);
			map.put(Constants.KeyStr.USER_NAME, topicName);
			if(topic){
				return true;
			}

		}catch (Exception e){
			log.error("delete topic has error,please check,",e);
		}
		return false;

	}

	/**
	 * @author Lion
	 * @description 判断是否存在
	 * @param  result
	 * @return boolean
	 */
	private boolean checkResult(Integer result) {
		return result > 0;
	}

	/**
	 * @author Lion
	 * @description 创建主题
	 * @param  topicInfo
	 * @return java.lang.Boolean
	 */
	public Boolean adminCreateTopic(TopicInfoEntity topicInfo) throws Exception {
		boolean flag = true;
		if(exitsTopic(topicInfo.getTopicName(),Long.parseLong(topicInfo.getClusterId()))){
			throw new Exception("Topic already exists in the ");
		}
		try {
			if(kafkaAdminCreateTopic(topicInfo)){
				TopicInfoEntity topicInfo1 = topicInfo;
				topicInfo1.setTtl(86400000L);

				if(!this.insert(topicInfo1)) {
					flag=false;
				}
			}
			return flag;
		}catch (Exception e){

		}
		return flag;
	}

	/**
	 * @author Lion
	 * @description 保存保存到数据库中
	 * @param  * @param topic
	 * @return boolean
	 */
	public boolean insert(TopicInfoEntity topic) {
		topic.setCreateTime(LocalDateTime.now());
		Integer result =  topicInfoMapper.insert(topic);
		return checkResult(result);
	}

	/**
	 * @author Lion
	 * @description 判断数据库是否存在
	 * @param  * @param topic
	 * @return boolean
	 */
	public boolean exitsTopic(String topic,Long clusterId) {
		Map<String,Object> columnMap = new HashMap<>();
		columnMap.put(Constants.KeyStr.TOPIC_NAME, topic);
		columnMap.put(Constants.KeyStr.CLUSTER_ID, clusterId);
		List<TopicInfoEntity> topicList = topicInfoMapper.selectByMap(columnMap);
		return !topicList.isEmpty();
	}

	/**
	 * @author Lion
	 * @description 创建
	 * @param  * @param topicInfo
	 * @return java.lang.Boolean
	 */
	public Boolean kafkaAdminCreateTopic(TopicInfoEntity topicInfo) throws Exception {
		if(!kafkaAdminService.getKafkaAdmins(topicInfo.getClusterId()).createTopicIfNotExists(topicInfo.getTopicName(), topicInfo.getPartition(),topicInfo.getReplication(), (topicInfo.getTtl().intValue() * 3600 * 1000))) {
			return Boolean.FALSE;
		}
		return Boolean.TRUE;
	}
   /**
    * @author Lion
    * @description
    * @param  * @param id
    * @return boolean
    */
	boolean deleteByClusterId(Long id) {
		Map<String,Object> map = new HashMap<>();
		map.put(Constants.KeyStr.CLUSTER_ID, id);
		if(selectByClusterId(map)) {
			return topicInfoMapper.deleteByMap(map) > 0;
		}
		return true;
	}
	/**
	 * @author Lion
	 * @description 在数据库中校验
	 * @param  map
	 * @return boolean
	 */
	private boolean selectByClusterId(Map<String, Object> map) {
		return !topicInfoMapper.selectByMap(map).isEmpty();
	}

	/**
	 * @author Lion
	 * @description 消费topic中的数据
	 * @param  * @param json
	 * @return com.alibaba.fastjson.JSONArray
	 */
    public JSONArray queryTopicData(Map<String,String> json){

		//获取请求参数
		Duration timeOut = Duration.ofMillis(Long.parseLong(json.get(Constants.KeyStr.WAIT_TIME)));
		boolean isCommit = Boolean.parseBoolean(json.get(Constants.KeyStr.IS_COMMIT));
		String clusterId = json.get(Constants.KeyStr.CLUSTERID);
		String groupID = json.get(Constants.KeyStr.GROUP_ID);
		String topicName = json.get(Constants.KeyStr.TOPICNAME);
		int dataSize = Integer.parseInt(json.get(Constants.KeyStr.RECORD_NUM));
		boolean isByPartition = Boolean.parseBoolean(json.get(Constants.KeyStr.ISBY_PARTITION));

		//获取数据
		int partition;
		long offset;
		ConsumerRecords<String,String> records;
		if (isByPartition){
			partition = Integer.parseInt(json.get(TopicConfig.PARTITION));
			offset = Long.parseLong(json.get(TopicConfig.OFFSET));
			records = kafkaConsumerService.consumer(clusterId,groupID,topicName,dataSize,timeOut,isCommit,partition,offset);
		}else {
			records = kafkaConsumerService.consumer(clusterId,groupID,topicName,dataSize,timeOut,isCommit);
		}

		//遍历封装数据
		JSONArray array = new JSONArray(dataSize + 1);
		for (ConsumerRecord<String, String> record : records){
			KafkaRecord kafkaRecord = new KafkaRecord();
			kafkaRecord.setPartition(record.partition());
			kafkaRecord.setOffset(record.offset());
			kafkaRecord.setKey(record.key());
			kafkaRecord.setValue(record.value());
			array.add(kafkaRecord);
		}
		return array;
	}

	public void queryTopicDataDD(Map<String,String> json){

		//获取请求参数
		Duration timeOut = Duration.ofSeconds(1);
		boolean isCommit = true;
		String clusterId = json.get(Constants.KeyStr.CLUSTERID);
		String groupID = json.get(Constants.KeyStr.GROUP_ID);
		String topicName = json.get(Constants.KeyStr.TOPIC_NAME);
		int dataSize = 10;
		boolean isByPartition = false;

		//获取数据
		int partition;
		long offset;
		if (isByPartition){
			partition = Integer.parseInt(json.get(TopicConfig.PARTITION));
			offset = Long.parseLong(json.get(TopicConfig.OFFSET));
			kafkaConsumerService.consumer(clusterId,groupID,topicName,dataSize,timeOut,isCommit,partition,offset);
		}else {
			kafkaConsumerService.consumerDD(clusterId,groupID,topicName,dataSize,timeOut,true);
		}
	}

	/**
	 * @author Lion
	 * @description
	 * @param clusterId 集群id
	 * @param topicName 主题名称
	 * @param key key
	 * @param value value
	 * @return void
	 */
	public void sendData(String clusterId,String topicName,String key,String value) throws ExecutionException, InterruptedException {

		ClusterEntity cluster = clusterMapper.selectById(clusterId);
		Properties properties = new Properties();
		properties.put("bootstrap.servers",cluster.getBroker());
		properties.put("key.serializer", StringSerializer.class);
		properties.put("value.serializer", StringSerializer.class);
		KafkaProducers<String, String> pro  = new KafkaProducers<>(properties);

		try {
			pro.send(topicName, key, value);
		} catch (Exception e) {
			log.error("  Producer  topic: {} error! key:{}  value:{} ", topicName,key, value, e);
			throw e;
		} finally {
			try {
				pro.close();
			} catch (IOException e) {
				log.error("close Producer error! ");
			}
		}
	}

}
